package com.ybliang.plat.common.mq.kafka.producer.configuration;

import com.ybliang.plat.common.mq.kafka.producer.configuration.properties.KafkaMultiProducerProperties;
import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
import org.springframework.core.env.Environment;
import org.springframework.core.type.AnnotationMetadata;

import java.util.Properties;

/**
 * @author liangyb1
 */
public class KafkaProducerImportBeanDefinitionRegistrar implements ImportBeanDefinitionRegistrar, EnvironmentAware {

    public static Logger LOGGER = LoggerFactory.getLogger(KafkaProducerImportBeanDefinitionRegistrar.class);

    private Environment environment;

    private KafkaMultiProducerProperties kafkaMultiProducers;

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = environment;
        kafkaMultiProducers = Binder.get(environment)
                .bind(KafkaMultiProducerProperties.PREFIX, KafkaMultiProducerProperties.class).get();
    }

    @Override
    public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
        if(null == kafkaMultiProducers || kafkaMultiProducers.getProducers().isEmpty()) {
            return ;
        }
        kafkaMultiProducers.getProducers().keySet().forEach(beanName -> registerBean(beanName, registry));
    }

    private void registerBean(String beanName, BeanDefinitionRegistry registry) {
        RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(Producer.class, () -> createKafkaProducer(beanName));
        registry.registerBeanDefinition(beanName, rootBeanDefinition);
        LOGGER.info("KafkaProducerImportBeanDefinitionRegistrar register bean {} done", beanName);
    }


    private <K, V> Producer<K, V> createKafkaProducer(String kafkaBeanName) {
        KafkaMultiProducerProperties.KafkaProducerProperties kafkaProducerProperties = kafkaMultiProducers.getProducers().get(kafkaBeanName);
        Properties props = new Properties();
        props.put("bootstrap.servers", kafkaProducerProperties.getBootstartServers());
        props.put("retries", kafkaProducerProperties.getRetries());
        props.put("batch.size", kafkaProducerProperties.getBatchSize());
        props.put("linger.ms", kafkaProducerProperties.getLingerMs());
        props.put("key.serializer", kafkaProducerProperties.getKeySerializer());
        props.put("value.serializer", kafkaProducerProperties.getValueSerializer());
        Producer<K, V> producer = new org.apache.kafka.clients.producer.KafkaProducer(props);
        return producer;
    };


}
